package cn.gwm.flink.streaming.strategy.cloudpredict;

import cn.gwm.flink.streaming.constant.ChargePredictFields;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.MapFunction;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class LongLat implements MapFunction<JSONObject, JSONObject> {
    public LongLat(List<String> lines){
        this.lines = lines;
        for (String line : lines) {
            try {
                String[] split = line.split(",",-1);
                String key = split[0].concat(split[1]);
                String value = split[2].concat(",").concat(split[3]);
                infoMap.put(key,value);
            }catch (Exception e){
                e.printStackTrace();
                System.out.println(line);
            }
        }
    }
    List<String> lines;
    Map<String,String> infoMap = new HashMap<>(16);
    @Override
    public JSONObject map(JSONObject line) throws Exception {
        try {
            String vin = line.getString(ChargePredictFields.ins().srcFieldVin);
            String startChargeTime = line.getString(ChargePredictFields.ins().srcFieldStartChargeTime);
            String s = infoMap.get(vin.concat(startChargeTime));
            if (StringUtils.isBlank(s)){
                line.put(ChargePredictFields.ins().srcFieldLongitude,ChargePredictFields.ins().defaultVal);
                line.put(ChargePredictFields.ins().srcFieldLatitude, ChargePredictFields.ins().defaultVal);
                return line;
            }
//            System.out.println("------>"+s);
            String[] split = s.split(",");
            line.put(ChargePredictFields.ins().srcFieldLongitude,split[0]);
            line.put(ChargePredictFields.ins().srcFieldLatitude, split[1]);
            return line;
        }catch (Exception e){
            e.printStackTrace();
        }
        return line;
    }
}
